package com.letv.dataprocess.cp_info;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.letv.bean.SummaryBean;
import com.letv.conf.ConfigFactory;
import com.letv.dataprocess.DataProcessInterface;
import com.letv.dataprocess.DataProcessResult;
import com.letv.download.DownloadUtils;

public class CpInfoDataProcess implements DataProcessInterface {

    /**
     * Logger
     */
    private static final Logger LOGGER = LoggerFactory.getLogger(CpInfoDataProcess.class);

    /**
     * String, thread pool base size path in config file
     */
    private static final String POOL_SIZE_PATH = "threadPool.basesize";

    /**
     * String, thread pool base size.program will use it when this parameter does't exist in config file
     */
    private static final int POOL_SIZE = 10;

    /**
     * String, the number of each thread will process in config file
     */
    private static final String PAGE_SIZE_PATH = "thread.pageSize";

    /**
     * String, the number of each thread will process.program will use it when this parameter does't exist in config file
     */
    private static final int PAGE_SIZE = 1000;

    /**
     * DownloadUtils, common method for download data from cms
     */
    private DownloadUtils downloadUtils = new DownloadUtils();

    /**
     * process data
     * @param workerName String, the worker's name to process data
     * @return SummaryBean, the result information of process
     */
    @Override
    public SummaryBean process(String workerName) {
        SummaryBean result = null;
        try {
            result = new SummaryBean();
            result.setErrorMsg("");
            LOGGER.info("begin to process " + workerName + "...");
            List<Integer> idRangeList = this.downloadUtils.getIdRangeList();
            if(null == idRangeList) {
                result.setErrorMsg(result.getErrorMsg() + "[cp_info fail to get id range]");
                return result;
            }
            LOGGER.info("success to get cp_info id rage,minId=" + idRangeList.get(0).intValue() + ",maxId=" + idRangeList.get(1).intValue());
            this.processInThreadPool(result, idRangeList.get(0).intValue() - 1, idRangeList.get(1).intValue());
        } catch(Exception e) {
            LOGGER.error("fail to process " + workerName + "...", e);
        }
        return result;
    }

    /**
     * process cp info by thread pool between id range
     * @param summary SummaryBean, cp info process infomation
     * @param beginId int
     * @param endId int
     */
    private void processInThreadPool(SummaryBean summary, int beginId, int endId) {
        if(null == summary || endId <= beginId) {
            LOGGER.error("cp_info fail to process by thread pool,method processInThreadPool gets illegal arguments");
            return;
        }
        LOGGER.info("cp info begin to process by thread pool,beginId=" + beginId + ",endId=" + endId);

        ExecutorService threadPool = null;
        try {
            threadPool = Executors.newFixedThreadPool(ConfigFactory.getInt(CpInfoDataProcess.POOL_SIZE_PATH, CpInfoDataProcess.POOL_SIZE));
        } catch(Exception e) {
            LOGGER.error("cp_info fail to process by thread pool,fail to create thread pool");
            return;
        }

        int pageSize = 0;
        int totalWorkerCount = 0;
        try {
            pageSize = ConfigFactory.getInt(CpInfoDataProcess.PAGE_SIZE_PATH, CpInfoDataProcess.PAGE_SIZE);
            totalWorkerCount = (endId - beginId) / pageSize + 1;
        } catch(Exception e) {
            LOGGER.error("cp_info fail to process by thread pool,fail to set pageSize or calculate totalPageCount");
            return;
        }
        LOGGER.info("each worker process " + pageSize + " data,total need " + totalWorkerCount + " workers");

        List<CpInfoDataProcessWorker> workerList = null;
        try {
            workerList = new ArrayList<CpInfoDataProcessWorker>();
            for(int i = 1; i <= totalWorkerCount; i++) {
                workerList.add(new CpInfoDataProcessWorker(i-1, beginId + pageSize * (i - 1), beginId + pageSize * i));
            }
        } catch(Exception e) {
            LOGGER.error("cp_info fail to process by thread pool,fail to create process worker list");
            return;
        }

        List<Future<DataProcessResult>> processResultList = null;
        try {
            processResultList = threadPool.invokeAll(workerList);
            LOGGER.info("======================= cp info data process Done! =======================");
            int allData = 0;
            int cbaseSuccess = 0;
            int cbaseFail = 0;
            for(int i = 0; i < processResultList.size(); i++) {
                try {
                    Future<DataProcessResult> future = processResultList.get(i);
                    if(null == future) {
                        continue;
                    }
                    DataProcessResult dataProcessResult = future.get();
                    allData = allData + dataProcessResult.getRecordsCnt();
                    cbaseSuccess = cbaseSuccess + dataProcessResult.getCbaseSuccess();
                    cbaseFail = cbaseFail + dataProcessResult.getCbasefail();
                } catch (Exception e1) {
                    LOGGER.error("cp_info fail to get thread pool result", e1);
                    continue;
                }
            }
            LOGGER.info("cp info all records number:" + allData);
            LOGGER.info("cp info put into cbase successly:" + cbaseSuccess);
            LOGGER.info("cp info put into cbase fail:" + cbaseFail);
        } catch (Exception e) {
            LOGGER.error("cp_info fail to process by thread pool,fail to invoke thread list", e);
            summary.setErrorMsg(summary.getErrorMsg() + "[cp_info fail to invoke thread list]");
        } finally {
            threadPool.shutdown();
            LOGGER.info("process cp_info 's threadPool shutdown!");
        }
    }
}
